package cn.itcast.up.model.matchtag

import java.util.Properties

import cn.itcast.up.model.bean.{HBaseMeta, TagRule}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 国籍标签开发
  */
object NationalModel {

  def main(args: Array[String]): Unit = {
    // 编写SparkSQL代码.
    //   1. 构建一个SparkSession的对象
    val spark: SparkSession = SparkSession.builder()
      .master("local[*]")
      .appName("NationalModel")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._
    //   2. 加载MySQL数据源,整个数据库中的表数据都可以获取到了.
    val url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val tableName = "tbl_basic_tag"
    val props = new Properties
    val mysqlSource: DataFrame = spark.read.jdbc(url, tableName, props)
    //   3. 获取tbl_basic_tag中的数据:
    //      1. 4级标签的数据:id,rule:数据源信息
    val meta: HBaseMeta = mysqlSource.select('id, 'rule)
      .where("id = 484")
      .map(row => {
        //获取row中的rule规则字符串
        //inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,nationality
        val ruleStr: String = row.getAs[String]("rule")
        val map: Map[String, String] = ruleStr.split("##")
          .map(kv => {
            val arr: Array[String] = kv.split("=")
            (arr(0), arr(1))
          }).toMap
        HBaseMeta(map)
      }).collect()(0)

    //      2. 5级标签的数据:id,rule: 匹配的时候使用.
    val fiveMap: Map[String, String] = mysqlSource.select('id, 'rule)
      .where("pid = 484")
      .map(row => {
        //获取5级规则的ID和rule
        val id: String = row.getAs[Long]("id").toString
        val rule: String = row.getAs[String]("rule")
        //将数据封装为元组返回
        (rule, id)
      }).collect().toMap

    println(fiveMap)
    //    Map(4 -> 488, 5 -> 489, 1 -> 485, 2 -> 486, 3 -> 487)

    //   4. 获取HBase中的数据
    //      1. 使用自定义数据源加载HBase中的数据.
    val hbaseSource: DataFrame = spark.read
      //自定义数据源的全类名.
      .format("cn.itcast.up.model.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, meta.hbaseTable)
      .option(HBaseMeta.FAMILY, meta.family)
      .option(HBaseMeta.SELECTFIELDS, meta.selectFields)
      .load()
    //   5. 使用5级规则数据和HBase中的源数据进行匹配,找到用户所属的国籍.

    val getTagId = udf((national: String) => {
      //national: 1/2/3/4/5
      //从5级标签中获取标签ID,如果有就直接拿,如果没有就强制获取5:其它
      fiveMap.getOrElse(national, fiveMap.getOrElse("5", ""))
    })
    val newDF: DataFrame = hbaseSource.select('id.as("userid"), getTagId('nationality).as("tagIds"))
    //    newDF.show()
    //+------+------+
    //|userid|tagIds|
    //+------+------+
    //|     1|   485|
    //|    10|   485|
    //|   100|   485|


    //   6. 先查询HBase中的历史标签数据
    val oldDF: DataFrame = spark.read
      //自定义数据源的全类名.
      .format("cn.itcast.up.model.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test32")
      .option(HBaseMeta.FAMILY, meta.family)
      .option(HBaseMeta.SELECTFIELDS, "userid,tagIds")
      .load()
    //   7. 使用本次的计算标签结果和HBase中的历史标签数据进行整合.(全连接)
     val joinResult: DataFrame = newDF.join(oldDF,newDF.col("userid") === oldDF.col("userid"),"full")

    //标签合并
    //定义自定义函数进行合并操作
    val mergeTag = udf((newTag: String, oldTag: String) => {
      var tagIds = ""
      //判断新老数据是否为空 isBlank ""
      val string = "" //不是null,是空
      val str1 = "                   " //是空
      if (StringUtils.isBlank(newTag)){
        tagIds = oldTag
      }
      if (StringUtils.isBlank(oldTag)){
        tagIds = newTag
      }
      //如果新老数据都不为空.那么就开始整合
      if (StringUtils.isNotBlank(newTag) && StringUtils.isNotBlank(oldTag)){
        val tmpTag: String = oldTag + "," + newTag
        //将数据使用,进行切割,之后转换为set去重,最后再转换为字符串
        tagIds = tmpTag.split(",").toSet.mkString(",")
      }
      //返回最终结果
      tagIds
    })
    //对标签结果数据进行选择(获取)
    val result: DataFrame = joinResult.select(
      //如果新数据用户ID不为空,就使用新数据的ID
      when(newDF.col("userid").isNotNull, newDF.col("userid"))
        .when(newDF.col("userid").isNull, oldDF.col("userid"))
        .as("userid"),
      mergeTag(newDF.col("tagIds"), oldDF.col("tagIds")).as("tagIds")
    )
    result.show()


    //   8. 将整合之后的结果存入HBase.
    //      1. 使用自定义数据源将表情结果存入HBase
    result.write
      .format("cn.itcast.up.model.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test32")
      .option(HBaseMeta.FAMILY, meta.family)
      .save()
  }
}
